package jmine.tec.subscription.impl;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.Map.Entry;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import jmine.tec.cache.ExpiringMap;
import jmine.tec.subscription.Publication;
import jmine.tec.subscription.PublicationListener;
import jmine.tec.subscription.PublicationStore;
import jmine.tec.subscription.Publisher;
import jmine.tec.subscription.PublisherManager;
import jmine.tec.subscription.Subscription;
import jmine.tec.subscription.SubscriptionSet;
import jmine.tec.subscription.SubscriptionSettings;
import jmine.tec.subscription.replicator.AbstractPublicationWrapper;
import jmine.tec.subscription.replicator.PublicationReplicator;
import jmine.tec.subscription.replicator.SerializerReplicator;
import jmine.tec.utils.debug.impl.DebugRuntimeFactory;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;

/**
 * Implementacao basica de {@link SubscriptionManagerImplementor}
 * 
 * @author takeshi
 */
public class SubscriptionManagerImpl implements SubscriptionManagerImplementor, DisposableBean, ApplicationListener {

    private static final int REFRESH_PERIOD = 300;

    private static final int DEFAULT_TIMEOUT_PERIOD = 30;

    private static final ExecutorService EXECUTOR = Executors.newSingleThreadExecutor(new ThreadFactory() {

        public Thread newThread(Runnable r) {
                Thread t = new Thread(r, "SubscriptionManagerThread");
                t.setDaemon(true);
                return t;
            }
    });

    private final Map<Long, SubscriptionSetImpl> subscriptions = new ConcurrentHashMap<Long, SubscriptionSetImpl>();

    private final List<ExpiringSubscriptionSetImpl> expiring = new LinkedList<ExpiringSubscriptionSetImpl>();

    private final ConcurrentMap<Publisher<?>, PublicationStore<?>> cacheStore = new ConcurrentHashMap<Publisher<?>, PublicationStore<?>>();

    private final ConcurrentMap<Publisher<?>, ConcurrentMap<SubscriptionSettings, Queue<PublicationListener<?>>>> allSubscriptions =
            new ConcurrentHashMap<Publisher<?>, ConcurrentMap<SubscriptionSettings, Queue<PublicationListener<?>>>>();

    private PlatformTransactionManager transactionManager;

    private PublicationReplicator replicator = new SerializerReplicator();

    private PublisherManager publisherManager;

    public static final TransactionDefinition DEFAULT_TX_DEFINITION =
            new DefaultTransactionDefinition(TransactionDefinition.PROPAGATION_SUPPORTS);

    /**
     * 
     */
    public SubscriptionManagerImpl() {
        DebugRuntimeFactory.getInstance().exposeAsJMX("subscriptionManager", this);
        ExpiringMap.ExecutorHolder.EXECUTOR.scheduleWithFixedDelay(new Runnable() {
                public void run() {
                    evictExpiredSubscriptionSets();
                }
        }, DEFAULT_TIMEOUT_PERIOD, DEFAULT_TIMEOUT_PERIOD, TimeUnit.SECONDS);
    }

    /**
     * Chama os callbacks para o publication passado. Caso o callback necessite de uma transacao, o callback eh agendado para ser executado
     * em uma thread separada, com uma transacao isolada e unica. Caso um replicator esteja configurado, a mensagem eh replicada.
     * 
     * @param source the {@link Publisher}
     * @param criteria the {@link SubscriptionSettings}
     * @param publication the {@link Publication}
     * @see PublicationListener#isTransactionRequired()
     * @param <P> o tipo concreto da publication
     */
    public <P extends Publication> void broadcast(final Publisher<P> source, final SubscriptionSettings criteria, final P publication) {
        this.localPublish(source, criteria, publication);
    }

    /**
     * Faz a invocacao dos callbacks
     * 
     * @param <P> o tipo da publication
     * @param source o Publisher
     * @param criteria os {@link SubscriptionSettings}
     * @param publication {@link Publication}
     */
    @SuppressWarnings("unchecked")
    private <P extends Publication> void invokeCallbacks(final Publisher<P> source, final SubscriptionSettings criteria, final P publication) {
        ConcurrentMap<SubscriptionSettings, Queue<PublicationListener<?>>> map = allSubscriptions.get(source);
        if (map != null) {
            Queue<PublicationListener<?>> list = map.get(criteria);
            if (list != null) {
                for (final PublicationListener publicationListener : list) {
                    invokeCallback(source, criteria, publication, publicationListener);
                }
            }
        }
    }

    /**
     * Invoca 1 callback, fazendo tratamento de assincronia e/ou transacao
     * 
     * @param <P> o tipo da {@link Publication}
     * @param source o source da {@link Publication}
     * @param criteria o {@link SubscriptionSettings}
     * @param publication o {@link Publication}
     * @param publicationListener o {@link PublicationListener}
     */
    private <P extends Publication> void invokeCallback(final Publisher<P> source, final SubscriptionSettings criteria,
            final P publication,
            final PublicationListener<P> publicationListener) {
        if (publicationListener.isTransactionRequired()) {
            // schedule the publish
            Runnable task = new Runnable() {
                public void run() {
                    synchronized (source) {
                        publicationListener.onPublication(source, criteria, publication);
                    }
                }
            };
            EXECUTOR.execute(this.newTransactedRunnable(task));
        } else if (publicationListener.isAssynchronous()) {
            Runnable task = new Runnable() {
                public void run() {
                    synchronized (source) {
                        publicationListener.onPublication(source, criteria, publication);
                    }
                }
            };
            EXECUTOR.execute(task);
        } else {
            try {
                publicationListener.onPublication(source, criteria, publication);
            } catch (Exception e) {
                e.printStackTrace();
                DebugRuntimeFactory.getInstance().getOrCreateMessageStore("subscriptionManager").postStackTrace(e);
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    public <P extends Publication> Subscription addSubscription(SubscriptionSet s, Publisher<P> source, SubscriptionSettings settings) {
        return this.addSubscription(s.getToken(), source, settings);
    }

    /**
     * Adiciona uma nova inscricao com o callback passado.
     * 
     * @param <P> o tipo concreto da publication
     * @param source {@link Publisher}
     * @param settings {@link SubscriptionSettings}
     * @param queueSubscriptionSet {@link SubscriptionSetImpl}
     * @param e {@link SubscriptionImpl}
     * @param publicationListener {@link PublicationListener}
     * @return a {@link Subscription} gerada (ou uma representacao)
     */
    private <P extends Publication> Subscription addSubscriptionInternal(final Publisher<P> source, final SubscriptionSettings settings,
            final SubscriptionSetImpl queueSubscriptionSet, SubscriptionImpl e, final PublicationListener<P> publicationListener) {
        final PublicationListener<P> wrapped = wrapIfNecessary(queueSubscriptionSet, publicationListener);
        queueSubscriptionSet.renewTimeout();
        queueSubscriptionSet.addSubscription(e, wrapped);
        final ConcurrentMap<SubscriptionSettings, Queue<PublicationListener<?>>> map = initMapForPublisher(source);
        final Queue<PublicationListener<?>> queue = getQueueFor(settings, map);
        final PublicationStore<P> store = getOrCreatePublicationStore(source);
        synchronized (source) {
            P lastPublication = store.lastPublication(settings);
            if (lastPublication == null) {
                lastPublication = source.newPublication(settings);
            }
            broadcast(source, settings, lastPublication);
            invokeCallback(source, settings, lastPublication, wrapped);
            queue.add(wrapped);
            return e;
        }
    }

    /**
     * Faz um wrap no publicationListener passado se necessario
     * 
     * @param <P> {@link Publication}
     * @param queueSubscriptionSet {@link SubscriptionSetImpl}
     * @param publicationListener {@link PublicationListener}
     * @return PublicationiListener
     */
    private <P extends Publication> PublicationListener<P> wrapIfNecessary(final SubscriptionSetImpl queueSubscriptionSet,
            final PublicationListener<P> publicationListener) {
        PublicationListener<P> wrapped =
                (queueSubscriptionSet instanceof ExpiringSubscriptionSetImpl) ? new RenewingPublicationListener<P>(queueSubscriptionSet,
                publicationListener) : publicationListener;
        return wrapped;
    }

    /**
     * Procura uma PublicationStore do cache ou cria uma nova, adicionando ao cache.
     * 
     * @param source o {@link Publisher}
     * @return {@link PublicationStore}
     * @param <P> o tipo concreto da publication
     */
    @SuppressWarnings("unchecked")
    private <P extends Publication> PublicationStore<P> getOrCreatePublicationStore(Publisher<P> source) {
        PublicationStore<P> store = (PublicationStore<P>) this.cacheStore.get(source);
        if (store == null) {
            store = new ExpiringPublicationStore<P>();
            PublicationStore<P> oldStore = (PublicationStore<P>) cacheStore.putIfAbsent(source, store);
            if (oldStore != null) {
                store = oldStore;
            }
            DebugRuntimeFactory.getInstance().exposeAsJMX("publicationStore,identification=" + source.identification(), store);
        }
        return store;
    }

    /**
     * Devolve uma fila para o {@link SubscriptionSettings} passado, inserindo no mapa passado, caso nenhum seja encontrado.
     * 
     * @param settings a chave para a file
     * @param map o {@link ConcurrentMap}
     * @return {@link Queue} - tipicamente um {@link ConcurrentLinkedQueue}
     * @see ConcurrentMap#putIfAbsent(Object, Object)
     */
    private Queue<PublicationListener<?>> getQueueFor(SubscriptionSettings settings,
            ConcurrentMap<SubscriptionSettings, Queue<PublicationListener<?>>> map) {
        Queue<PublicationListener<?>> newQueue = new ConcurrentLinkedQueue<PublicationListener<?>>();
        Queue<PublicationListener<?>> oldQueue = map.putIfAbsent(settings, newQueue);
        return oldQueue == null ? newQueue : oldQueue;
    }

    /**
     * Inicializa um mapa para um {@link Publisher} especifico, se nao existe nenhum el {@link #allSubscriptions}.
     * 
     * @param source o {@link Publisher}
     * @return {@link ConcurrentMap}
     */
    private ConcurrentMap<SubscriptionSettings, Queue<PublicationListener<?>>> initMapForPublisher(Publisher<?> source) {
        ConcurrentMap<SubscriptionSettings, Queue<PublicationListener<?>>> newMap =
                new ConcurrentHashMap<SubscriptionSettings, Queue<PublicationListener<?>>>();
        ConcurrentMap<SubscriptionSettings, Queue<PublicationListener<?>>> oldMap = this.allSubscriptions.putIfAbsent(source, newMap);
        return oldMap == null ? newMap : oldMap;
    }

    /**
     * {@inheritDoc}
     */
    public void cancelSubscription(SubscriptionSet su) {
        SubscriptionSetImpl queueSubscriptionSet = this.subscriptions.remove(su.getToken());
        if (queueSubscriptionSet != null) {
            doCancel(queueSubscriptionSet);
        }
    }

    /**
     * cancela a subscriptionset
     * 
     * @param queueSubscriptionSet {@link SubscriptionSetImpl}
     */
    private void doCancel(SubscriptionSetImpl queueSubscriptionSet) {
        Map<Subscription, PublicationListener<?>> map = queueSubscriptionSet.getManagedSubscriptions();
        Set<Entry<Subscription, PublicationListener<?>>> set = map.entrySet();
        for (Entry<Subscription, PublicationListener<?>> entry : set) {
            this.clearCallback(entry.getKey().getPublisher(), entry.getKey(), entry.getValue());
        }
        queueSubscriptionSet.removeAll();
        this.subscriptions.remove(queueSubscriptionSet.getToken());
    }

    /**
     * {@inheritDoc}
     */
    public Map<Subscription, Publication> getChanges(SubscriptionSet set) {
        final SubscriptionSetImpl ss = this.subscriptions.get(set.getToken());
        if (ss == null) {
            return null;
        }
        return ss.getPublications();
    }

    /**
     * {@inheritDoc}
     */
    public SubscriptionSet getSubscriptionSet(long token) {
        SubscriptionSetImpl queueSubscriptionSet = this.subscriptions.get(token);
        if (queueSubscriptionSet != null) {
            queueSubscriptionSet.renewTimeout();
            return queueSubscriptionSet.toRepresentation();
        }
        return null;
    }

    /**
     * {@inheritDoc}
     */
    public SubscriptionSet newSubscription() {
        SubscriptionSetImpl impl = new SubscriptionSetImpl();
        this.subscriptions.put(impl.getToken(), impl);
        return impl.toRepresentation();
    }

    /**
     * {@inheritDoc}
     */
    public void unsubscribe(SubscriptionSet s, Publisher<?> source, SubscriptionSettings settings) {
        long token = s.getToken();
        final SubscriptionSetImpl subscriptionSetImpl = this.subscriptions.get(token);
        if (subscriptionSetImpl == null) {
            return;
        }
        subscriptionSetImpl.renewTimeout();
        Subscription sub = subscriptionSetImpl.findSubscription(source, settings);
        if (sub != null) {
            PublicationListener<?> associatedCallback = subscriptionSetImpl.getCallback(sub);
            clearCallback(source, sub, associatedCallback);
            subscriptionSetImpl.removeSubscription(sub);
        }
    }

    /**
     * Apaga todos os callbacks associados
     * 
     * @param source {@link Publisher}
     * @param selected {@link Subscription}
     * @param publicationListener o {@link PublicationListener} a ser removido
     */
    private void clearCallback(Publisher<?> source, Subscription selected, PublicationListener<?> publicationListener) {
        ConcurrentMap<SubscriptionSettings, Queue<PublicationListener<?>>> set = this.allSubscriptions.get(source);
        if (set != null) {
            Queue<PublicationListener<?>> queue = set.get(selected.getSettings());
            if (queue != null) {
                if (queue.remove(publicationListener)) {
                    if (queue.isEmpty()) {
                        PublicationStore<?> publicationStore = this.cacheStore.get(source);
                        source.evict(selected.getSettings());
                        if (publicationStore != null) {
                            publicationStore.evict(selected.getSettings());
                        }
                    }
                }
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    public boolean hasSubscription(Publisher<?> source, SubscriptionSettings settings) {
        ConcurrentMap<SubscriptionSettings, Queue<PublicationListener<?>>> set = this.allSubscriptions.get(source);
        if (set == null) {
            return false;
        }
        return set.containsKey(settings);
    }

    /**
     * {@inheritDoc}
     */
    public void destroy() {
        Collection<SubscriptionSetImpl> values = new ArrayList<SubscriptionSetImpl>(this.subscriptions.values());
        for (SubscriptionSetImpl subscriptionSetImpl : values) {
            doCancel(subscriptionSetImpl);
        }
        this.subscriptions.clear();
        for(PublicationStore<?> store: this.cacheStore.values()) {
            store.evictAll();
        }
        this.evictCacheStore();
        if(this.refreshFuture != null) {
            this.refreshFuture.cancel(true);
            ExpiringMap.ExecutorHolder.EXECUTOR.purge();
        }
    }

    /**
     * {@inheritDoc}
     */
    public <P extends Publication> Subscription addSubscription(long s, Publisher<P> source, SubscriptionSettings settings) {
        final SubscriptionSetImpl queueSubscriptionSet = this.subscriptions.get(s);
        final SubscriptionImpl e = queueSubscriptionSet.newSubscription(settings, source);
        return addSubscriptionInternal(source, settings, queueSubscriptionSet, e, new PushPublicationListener<P>(e, queueSubscriptionSet));
    }

    /**
     * {@inheritDoc}
     */
    public <P extends Publication> Subscription addSubscription(SubscriptionSet s, Publisher<P> source, SubscriptionSettings settings,
            PublicationListener<P> listener) {
        return this.addSubscription(s.getToken(), source, settings, listener);
    }

    /**
     * {@inheritDoc}
     */
    public <P extends Publication> Subscription addSubscription(long s, Publisher<P> source, SubscriptionSettings settings,
            PublicationListener<P> listener) {
        final SubscriptionSetImpl queueSubscriptionSet = this.subscriptions.get(s);
        final SubscriptionImpl e = queueSubscriptionSet.newSubscription(settings, source);
        return addSubscriptionInternal(source, settings, queueSubscriptionSet, e, listener);
    }

    /**
     * {@inheritDoc}
     */
    public SubscriptionSet newExpiringSubscription(long timeout, TimeUnit unit) {
        final ExpiringSubscriptionSetImpl impl = new ExpiringSubscriptionSetImpl(unit.toMillis(timeout));
        synchronized (expiring) {
            this.expiring.add(impl);
        }
        this.subscriptions.put(impl.getToken(), impl);
        return impl.toRepresentation();
    }

    /**
     * Faz um wrap envolta do {@link Runnable} passado para que seja executado em um contexto transacional. Usa como definicao de transacao
     * o {@link #DEFAULT_TX_DEFINITION}
     * 
     * @param task {@link Runnable}
     * @return {@link Runnable}
     */
    private Runnable newTransactedRunnable(final Runnable task) {
        return new Runnable() {
            public void run() {
                TransactionStatus status = transactionManager.getTransaction(DEFAULT_TX_DEFINITION);
                try {
                    task.run();
                    transactionManager.commit(status);
                } finally {
                    if (!status.isCompleted()) {
                        transactionManager.rollback(status);
                    }
                }
            }
        };
    }

    /**
     * @param transactionManager the transactionManager to set
     */
    public void setTransactionManager(PlatformTransactionManager transactionManager) {
        this.transactionManager = transactionManager;
    }

    /**
     * @param replicator the replicator to set
     */
    public void setReplicator(PublicationReplicator replicator) {
        this.replicator = replicator;
        if (replicator != null) {
            replicator.setSubscriptionManager(this);
        }
    }

    /**
     * {@inheritDoc}
     */
    @SuppressWarnings("unchecked")
    public <P extends Publication> P getPublication(Publisher<P> source, SubscriptionSettings settings) {
        PublicationStore<P> store = (PublicationStore<P>) this.cacheStore.get(source);
        if (store == null) {
            return null;
        }
        return store.lastPublication(settings);
    }

    /**
     * {@inheritDoc}
     */
    public <P extends Publication> void broadcastPartialPublication(Publisher<P> source, SubscriptionSettings settings,
            PartialPublication<P> published) {
        if (this.replicator != null) {
            this.replicator.synchronizePartialPublication(source, settings, published, this);
        }
        this.localPublishPartialPublication(source, settings, published);
    }

    /**
     * 'Traduz' um {@link Publisher} para um que esta 'cached'
     * 
     * @param <P> o tipo da {@link Publication}
     * @param publisher o {@link Publisher}
     * @return P
     */
    @SuppressWarnings("unchecked")
    protected <P extends Publication> Publisher<P> translatePublisher(Publisher<P> publisher) {
        return (Publisher<P>) this.publisherManager.getPublisher(publisher.identification());
    }

    /**
     * {@inheritDoc}
     */
    public <P extends Publication> void localPublish(Publisher<P> source, SubscriptionSettings settings, P published) {
        Publisher<P> translated = this.translatePublisher(source);
        synchronized (translated) {
            final PublicationStore<P> store = getOrCreatePublicationStore(translated);
            if (store.update(settings, published)) {
                invokeCallbacks(translated, settings, published);
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    public <P extends Publication> void localPublishPartialPublication(Publisher<P> source, SubscriptionSettings settings,
            PartialPublication<P> published) {
        Publisher<P> translated = this.translatePublisher(source);
        final PublicationStore<P> store = (PublicationStore<P>) this.getOrCreatePublicationStore(translated);
        final P merged;
        final P last;
        synchronized (translated) {
            last = store.lastPublication(settings);
            if (last != null) {
                merged = published.merge(last);
            } else if (hasSubscription(translated, settings)) {
                merged = translated.newPublication(settings);
            } else {
                return;
            }
            if (store.update(settings, merged)) {
                this.invokeCallbacks(translated, settings, merged);
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    public void onApplicationEvent(ApplicationEvent event) {
        if (event instanceof AbstractPublicationWrapper) {
            AbstractPublicationWrapper wrapper = (AbstractPublicationWrapper) event;
            wrapper.broadcast(this, this.publisherManager);
        }
    }

    /**
     * @param publisherManager the publisherManager to set
     */
    public void setPublisherManager(PublisherManager publisherManager) {
        this.publisherManager = publisherManager;
    }

    /**
     * {@inheritDoc}
     */
    public Publication getChanges(SubscriptionSet set, long subscriptionToken) {
        final SubscriptionSetImpl ss = this.subscriptions.get(set.getToken());
        if (ss == null) {
            return null;
        }
        return ss.getPublication(subscriptionToken);
    }

    /**
     * {@inheritDoc}
     */
    public void unsubscribe(SubscriptionSet set, long subscriptionToken) {
        SubscriptionSetImpl setImpl = this.subscriptions.get(set.getToken());
        if (setImpl == null) {
            return;
        }
        Subscription subscription = setImpl.getSubscription(subscriptionToken);
        if (subscription == null) {
            return;
        }
        this.unsubscribe(set, subscription.getPublisher(), subscription.getSettings());
    }

    /**
     * @return int
     */
    public int getSubscribers() {
        return this.subscriptions.size();
    }

    /**
     * 
     */
    public void evictCacheStore() {
        Collection<PublicationStore<?>> values = cacheStore.values();
        for (PublicationStore<?> publicationStore : values) {
            publicationStore.evictAll();
        }
    }

    /**
     * @return int
     */
    public final int evictExpiredSubscriptionSets() {
        List<ExpiringSubscriptionSetImpl> expired = new LinkedList<ExpiringSubscriptionSetImpl>();
        synchronized (expiring) {
            for (ExpiringSubscriptionSetImpl expiringSubscriptionSetImpl : this.expiring) {
                if (expiringSubscriptionSetImpl.isExpired()) {
                    expired.add(expiringSubscriptionSetImpl);
                }
            }
        }
        for (ExpiringSubscriptionSetImpl expiringSubscriptionSetImpl : expired) {
            this.doCancel(expiringSubscriptionSetImpl);
        }
        synchronized (expiring) {
            expiring.removeAll(expired);
        }
        return expired.size();
    }

    /**
     * @return int
     */
    public final int countCallbacks() {
        Collection<ConcurrentMap<SubscriptionSettings, Queue<PublicationListener<?>>>> values = this.allSubscriptions.values();
        int t = 0;
        for (ConcurrentMap<SubscriptionSettings, Queue<PublicationListener<?>>> concurrentMap : values) {
            Collection<Queue<PublicationListener<?>>> collection = concurrentMap.values();
            for (Queue<PublicationListener<?>> queue : collection) {
                t += queue.size();
            }
        }
        return t;
    }

    /**
     * {@inheritDoc}
     */
    public <P extends Publication> Future<Subscription> addLazySubscription(final long subscriptionSet, final Publisher<P> source,
            final SubscriptionSettings settings) {
        Callable<Subscription> callable = new Callable<Subscription>() {
            public Subscription call() throws Exception {
                return addSubscription(subscriptionSet, source, settings);
            }
        };
        return EXECUTOR.submit(callable);
    }

    /**
     * {@inheritDoc}
     */
    public <P extends Publication> Future<Subscription> addLazySubscription(final long subscriptionSet, final Publisher<P> source,
            final SubscriptionSettings settings, final PublicationListener<P> listener) {
        Callable<Subscription> callable = new Callable<Subscription>() {
            public Subscription call() throws Exception {
                return addSubscription(subscriptionSet, source, settings, listener);
            }
        };
        return EXECUTOR.submit(callable);
    }

    private ScheduledFuture<?> refreshFuture;

    /**
     * Quando o parametro passado for <code>true</code>, vai forcar uma reatualizacao de todas as subscriptions disponiveis a cada 5 minutos
     * 
     * @param v boolean
     */
    public synchronized void setAutoRefresh(boolean v) {
        if (v && this.refreshFuture == null) {
            this.refreshFuture= ExpiringMap.ExecutorHolder.EXECUTOR.scheduleWithFixedDelay(new Runnable() {
                    public void run() {
                        Set<SubscriptionSettings> done = new HashSet<SubscriptionSettings>();
                        for (SubscriptionSetImpl ssi : subscriptions.values()) {
                            if (!ssi.isExpired()) {
                                for (Subscription sub : ssi.getSubscriptions()) {
                                    if (done.add(sub.getSettings())) {
                                        refresh(sub.getPublisher(), sub.getSettings());
                                    }
                                }
                            }
                        }
                    }
            }, 0, REFRESH_PERIOD, TimeUnit.SECONDS);
        } else if (!v && refreshFuture != null) {
            refreshFuture.cancel(true);
        }
    }

    /**
     * @return boolean
     */
    public synchronized boolean isAutoRefresh() {
        return this.refreshFuture != null;
    }
    /**
     * Forca a atualizacao de uma publicacao na thread atual 
     * @param <E> o tipo da publicacao
     * @param publisher o publisher
     * @param settings o criterio
     */
    private <E extends Publication> void refresh(final Publisher<E> publisher, final SubscriptionSettings settings) {
        E publication = publisher.newPublication(settings);
        this.broadcast(publisher, settings, publication);
    }

}
